package org.aceor.funny.common.executor;

import com.google.common.collect.Sets;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 多线程执行框架.
 */
public final class ExecutorEngine {
    private static final Logger log = LoggerFactory.getLogger(ExecutorEngine.class);

    private final ListeningExecutorService executorService;

    private static final int IDLE_THREAD_NUM = 0;
    private static final int WAIT_TIME = 60;//second

    private static final int MAX_THREAD_NUM = Runtime.getRuntime().availableProcessors() * 2;

    public ExecutorEngine() {
        executorService = MoreExecutors.listeningDecorator(new ThreadPoolExecutor(IDLE_THREAD_NUM, MAX_THREAD_NUM,
                WAIT_TIME, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * 多线程执行任务.
     *
     * @param inputs      输入参数
     * @param executeUnit 执行单元
     * @param <I>         入参类型
     * @param <O>         出参类型
     * @return 执行结果
     */
    public <I, O> List<O> execute(final Iterable<I> inputs, final ExecuteUnit<I, O> executeUnit) {
        ListenableFuture<List<O>> futures = submitFutures(inputs, executeUnit);
        addCallback(futures);
        return getFutureResults(futures);
    }

    /**
     * 多线程执行任务并归并结果.
     *
     * @param inputs      执行入参
     * @param executeUnit 执行单元
     * @param mergeUnit   合并结果单元
     * @param <I>         入参类型
     * @param <M>         中间结果类型
     * @param <O>         最终结果类型
     * @return 执行结果
     */
    public <I, M, O> O execute(final Iterable<I> inputs, final ExecuteUnit<I, M> executeUnit, final MergeUnit<M, O> mergeUnit) {
        return mergeUnit.merge(execute(inputs, executeUnit));
    }

    private <I, O> ListenableFuture<List<O>> submitFutures(final Iterable<I> inputs, final ExecuteUnit<I, O> executeUnit) {
        Set<ListenableFuture<O>> result = Sets.newHashSet();
        for (final I each : inputs) {
            result.add(executorService.submit(new Callable<O>() {

                @Override
                public O call() throws Exception {
                    return executeUnit.execute(each);
                }
            }));
        }
        return Futures.allAsList(result);
    }

    private <T> void addCallback(final ListenableFuture<T> allFutures) {
        Futures.addCallback(allFutures, new FutureCallback<T>() {
            @Override
            public void onSuccess(final T result) {
                log.trace("Concurrent execute result success {}", result);
            }

            @Override
            public void onFailure(final Throwable thrown) {
                log.error("Concurrent execute result error {}", thrown);
            }
        });
    }

    private <O> O getFutureResults(final ListenableFuture<O> futures) {
        try {
            return futures.get();
        } catch (final InterruptedException | ExecutionException ex) {
            throw new RuntimeException(ex);
        }
    }
}
